package com.easychat.websocket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.easychat.constants.Constants;
import com.easychat.entity.config.AppConfig;
import com.easychat.entity.dto.MessageSendDTO;
import com.easychat.redis.RedisComponent;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.api.listener.MessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.LinkedList;
import java.util.UUID;

//@Slf4j
@Component("messageHandler")
public class MessageHandler {

    private static final Logger logger = LoggerFactory.getLogger(MessageHandler.class);

//    private static final String MESSAGE_TOPIC = "message.topic";
//    @Resource
//    private RedissonClient redissonClient;
    @Resource
    private ChannelContextUtils channelContextUtils;
    @Resource
    private RedisComponent redisComponent;
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private AppConfig appConfig;

    /**
     * 将消息投递到联系人对应的服务器
     * redis维护联系人和对应的服务器端口号
     * 由服务器端口号构造路由Key
     * @param messageSendDTO
     */
    public void sendMessage(MessageSendDTO messageSendDTO) {
        //获取联系人ID
        String contactId = messageSendDTO.getContactId();
        //获取联系人所连接的服务器端口号
        String targetServerPort = redisComponent.getServerPort(contactId);
        if (targetServerPort == null) {
            logger.info("找不到联系人所在服务器端口号，联系人：{}", contactId);
            return;
        }
        //说明联系人就连接在本服务器上，可直接发送
        if (targetServerPort.equals(appConfig.getServerPort().toString())) {
            channelContextUtils.sendMessageToUserOrGroup(messageSendDTO);
            return;
        }
        //根据联系人所在服务器端口号，获得路由key
        String routingKey = "key." + targetServerPort;
        //发送消息，注意序列化
        rabbitTemplate.convertAndSend(Constants.EXCHANGE_DIRECT, routingKey, JSONObject.toJSONString(messageSendDTO));
    }

    /**
     * 消费者
     * 接收消息推送
     * 在服务启动时，就通过路由Key（本服务器端口号）将自己的队列绑定到交换机
     * 注意不能采用System.getProperty("server.port")获取端口号，业务@Queue
     * 传入的必须是常量
     * 改用配置文件的方式，在editConfig中设置allowMultiInstances
     * 每启动一个服务之前，改端口号
     */
    @RabbitListener(bindings = @QueueBinding(
            // 配置队列信息：durable 设置为 true 表示队列持久化；autoDelete 设置为 false 表示关闭自动删除
            value = @Queue(value = Constants.QUEUE_NAME + "${server.port}", durable = "true", autoDelete = "false"),
            // 配置交换机信息：durable 设置为 true 表示队列持久化；autoDelete 设置为 false 表示关闭自动删除
            exchange = @Exchange(value = Constants.EXCHANGE_DIRECT, durable = "true", autoDelete = "false"),
            key = {Constants.ROUTING_KEY + "${server.port}"}
    ))
    public void listenMessage(String dateString, Message message, Channel channel) throws IOException {

        // 1、获取当前消息的 deliveryTag 值备用
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            // 2、正常业务操作
            logger.info("收到消息：{}", dateString);
            MessageSendDTO messageSendDTO = JSON.parseObject(dateString, MessageSendDTO.class);
            channelContextUtils.sendMessageToUserOrGroup(messageSendDTO);
            // 3、给 RabbitMQ 服务器返回 ACK 确认信息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            logger.error("***********错误信息：{}", e.getMessage());
            // 4、获取信息，看当前消息是否曾经被投递过
            Boolean redelivered = message.getMessageProperties().getRedelivered();
            if (!redelivered) {
                logger.error("尝试重新投递：{}", new String(message.getBody()));
                // 5、如果没有被投递过，那就重新放回队列，重新投递，再试一次
                channel.basicNack(deliveryTag, false, true);
            } else {
                logger.error("消息被拒绝：{}", new String(message.getBody()));
                // 6、如果已经被投递过，且这一次仍然进入了 catch 块，那么返回拒绝且不再放回队列
                channel.basicReject(deliveryTag, false);
            }
        }
    }

    /**
     * 服务启动时执行此方法，监听订阅的topic，一旦某一台服务器发布消息，
     * 订阅该topic的所有服务器都能收到
     * 消费者必须先订阅队列，生产者才能发布消息，否则消息会丢失。
     * PostConstruct注解执行时机：bean实例化 -> bean属性填充 -> 找到@PostConstruct注解（postProcessBeforeInitialization方法） ->
     * 执行@PostConstruct注解的方法（invokeInitMethods方法方法）
     */
//    @PostConstruct
//    public void listenMessage() {
////        Returns topic instance by name
//        RTopic topic = redissonClient.getTopic(MESSAGE_TOPIC);
////        Subscribes to this topic. MessageListener.onMessage is called when any message is published on this topic.
////        Params:
////        type – - type of message
////        listener – for messages
////        Returns:
////        locally unique listener id
////        <M> int addListener(Class<M> type, MessageListener<? extends M> listener);
//        topic.addListener(MessageSendDTO.class, new MessageListener<MessageSendDTO>() {
//            @Override
//            public void onMessage(CharSequence charSequence, MessageSendDTO messageSendDTO) {
//                //log.info("收到广播消息：{}", JSONObject.toJSONString(messageSendDTO));
//                channelContextUtils.sendMessageToUserOrGroup(messageSendDTO);
//            }
//        });
//    }

    /**
     * 发布订阅模式实现集群发信息
     * 某一台服务器发布消息
     * @param messageSendDTO
     */
//    public void sendMessage(MessageSendDTO messageSendDTO) {
//        RTopic topic = redissonClient.getTopic(MESSAGE_TOPIC);
////        Publish the message to all subscribers of this topic
////        Params:
////        message – to send
////        Returns:
////        the number of clients that received the message
////        long publish(Object message);
//        topic.publish(messageSendDTO);
//    }
}
